#pragma once

#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <memory>
#include <chrono>
#include <cassert>

#include "config/cfg.h"
#include "common/threadpool.h"
#include "event/event.h"
#include "common/log.h"


namespace evm {

using namespace std;
using namespace evm::common;
using namespace evm::config;
using namespace evm::event;

struct Event {
    bool wait(int timeout = -1) {
        unique_lock<mutex> lk(m_);

        if (timeout < 0) {
            cv_.wait(lk);
            return true;
        }
        else
            return cv_.wait_for(lk, chrono::seconds(timeout)) == cv_status::no_timeout;
    }

    void notify() {
        cv_.notify_one();
    }

    mutex m_;
    condition_variable cv_;
};

using Task = function<void ()>;

const int DEFAULT_INTERVAL = 60;

/* Service object for binaries running on hosts. */
struct Service {
    Service() : loop_(new EventLoop) {}

    /* Start a service. */
    virtual void start() {
        // Start the event loop in another thread
        loop_->start();
    }

    /* Stop a service. */
    virtual void stop() {
        loop_->stop();

        done_.notify();
    }

    /* Wait for a service to shut down. */
    bool wait(int timeout = -1) {
        return done_.wait(timeout);
    }

    void add_periodic_task(Task task, int interval, unsigned initial_delay = 0) {
        if (interval < 0)  // disable task
            return;

        if (interval == 0)  // use default interval
            interval = DEFAULT_INTERVAL;

        auto id = loop_->add_timer(static_cast<double>(initial_delay), [=] (size_t) {
            auto tic = time();
            log_info("periodic_task");
            task();
            log_info("periodic_task");
            auto toc = time();
            double next_interval = static_cast<double>(interval) - (toc - tic) / 1000;
            return next_interval > 0 ? next_interval : 0.;
        });
        log_info("add_periodic_task, time event {}, initial_delay {}, interval {}, {}", id, initial_delay, interval, reinterpret_cast<uintptr_t>(loop_.get()));
    }

    void add_oneshot_task(Task task, unsigned initial_delay = 0) {
        loop_->add_timer(static_cast<double>(initial_delay), [=] (size_t) {
            task();
            return EV_NOMORE;
        });
    }

protected:
    EventLoopPtr loop_;
    Event done_;
};

using ServicePtr = shared_ptr<Service>;

struct Services {
    void add(ServicePtr service) {
        // Add a service to a list
        services_.push_back(service);
        // Create a thread to run it
        threads_.emplace_back([service, this] {
            service->start();
            done_.wait();
        });
    }

    void stop() {
        // Gracefully shutdown services
        for (auto service: services_)
            service->stop();
        done_.notify();
    }

    void wait() {
        // Wait for services to shutdown
        for (auto service: services_)
            service->wait();

        // Wait for threads to exit
        for (auto& thread: threads_)
            thread.join();
        // Clear threads
        threads_.clear();
    }

    void restart() {
        // Stop services
        stop();

        // Start services in new threads
        for (auto service: services_)
            threads_.emplace_back([service, this] {
                service->start();
                done_.wait();
            });
    }

private:
    vector<ServicePtr> services_;
    vector<thread> threads_;  // one thread for one service
    Event done_;
};

using ServicesPtr = shared_ptr<Services>;

/* Launch one or more services and wait for them to complete. */
struct Launcher {
    Launcher(ConfigOptsPtr conf) : conf_(conf) {}

    /* Load and start the given service. */
    void launch_service(ServicePtr service) {
        services_->add(service);
    }

    /* Stop all services which are currently running. */
    void stop() {
        services_->stop();
    }

    /* Wait until all services have been stopped, and then return. */
    void wait() {
        services_->wait();
    }

    /* Reload config files and restart service. */
    void restart() {
        // TODO: reload config files
        services_->restart();
    }

private:
    ConfigOptsPtr conf_;
    ServicesPtr services_;
};

using LauncherPtr = shared_ptr<Launcher>;

/* Runs one or more service in a parent process. */
struct ServiceLauncher : Launcher {
    ServiceLauncher(ConfigOptsPtr conf) : Launcher(conf) {}
};

/* Launch a service with a given number of workers. */
struct ProcessLauncher : Launcher {
    ProcessLauncher(ConfigOptsPtr conf) : Launcher(conf) {}
};

inline LauncherPtr launch(ConfigOptsPtr conf, ServicePtr service, size_t workers = 1) {
    assert(workers >= 1);
    LauncherPtr launcher;
    if (workers == 1)
        launcher = LauncherPtr(new ServiceLauncher(conf));
    else
        launcher = LauncherPtr(new ProcessLauncher(conf));
    launcher->launch_service(service);
    return launcher;
}

}
